package com.qupeng.demo.kafka.kafkaspringbootproducer.admin;

import com.qupeng.demo.kafka.kafkaspringbootproducer.admin.model.ConfigOperation;
import com.qupeng.demo.kafka.kafkaspringbootproducer.admin.model.Configuration;
import com.qupeng.demo.kafka.kafkaspringbootproducer.config.KafkaProperties;
import com.qupeng.demo.kafka.kafkaspringbootproducer.model.CommonResponse;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.web.bind.annotation.*;

import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

@RestController
@RequestMapping("/admin/sync/config")
public class ConfigSyncAdminController {

    private static final Logger LOGGER = LoggerFactory.getLogger(ConfigSyncAdminController.class);

    @Autowired
    private KafkaAdmin kafkaAdmin;

    @Autowired
    private AdminClient adminClient;

    @Autowired
    private KafkaProperties kafkaProperties;

    @GetMapping("/topics/{name}")
    public CommonResponse<List<Configuration>> topicConfig(@PathVariable String name) {
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, name);
        DescribeConfigsOptions describeConfigsOptions = new DescribeConfigsOptions();
        describeConfigsOptions.timeoutMs(5000);
        DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Collections.singletonList(configResource), describeConfigsOptions);
        try {
            Config configResult = describeConfigsResult.all().get().get(configResource);

            List<Configuration> configs = configResult.entries().stream().map(configEntry -> {
                Configuration config = new Configuration();
                BeanUtils.copyProperties(configEntry, config);
                return config;
            }).collect(Collectors.toList());
            return new CommonResponse<>("kafka-admin", configs);
        } catch (InterruptedException | ExecutionException ex) {
            return new CommonResponse<>("admin-config-001", ex.getMessage(), "kafka-admin");
        }
    }

    @PutMapping("/topics/{name}")
    public CommonResponse<List<Configuration>> updateTopicConfig(@PathVariable String name, @RequestBody ConfigOperation operation) {
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, name);
        DescribeConfigsOptions describeConfigsOptions = new DescribeConfigsOptions();
        describeConfigsOptions.timeoutMs(5000);
        DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Collections.singletonList(configResource), describeConfigsOptions);
        try {
            Config configResult = describeConfigsResult.all().get().get(configResource);
            //TopicConfig.CLEANUP_POLICY_COMPACT
            ConfigEntry compaction = new ConfigEntry(operation.getName(), operation.getValue());
            if (!configResult.entries().contains(compaction)) {
                Collection<AlterConfigOp> configOp = new ArrayList<>();
                configOp.add(new AlterConfigOp(compaction, AlterConfigOp.OpType.forId(operation.getOperationType())));
                Map<ConfigResource, Collection<AlterConfigOp>> alterConf = new HashMap<>();
                alterConf.put(configResource, configOp);
                adminClient.incrementalAlterConfigs(alterConf).all().get();
            }
            configResult = describeConfigsResult.all().get().get(configResource);
            List<Configuration> configs = configResult.entries().stream().filter(configEntry -> configEntry.name().equals(operation.getName())).map(configEntry -> {
                Configuration config = new Configuration();
                BeanUtils.copyProperties(configEntry, config);
                return config;
            }).collect(Collectors.toList());
            return new CommonResponse<>("kafka-admin", configs);
        } catch (InterruptedException | ExecutionException ex) {
            return new CommonResponse<>("admin-config-001", ex.getMessage(), "kafka-admin");
        }
    }
}
